package com.ben.multiple.websocket;

import com.ben.multiple.websocket.entity.BizMessage;
import com.ben.multiple.websocket.entity.SysUser;
import com.ben.multiple.websocket.entity.UserTypeEnum;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * ServerEndpoint：将目前的类定义成一个websocket服务器端，注解的值将被用于监听用户连接的终端访问URL地址，
 * 客户端可以通过这个URL来连接到WebSocket服务器端
 *
 * @author Ben
 * @date 2020/3/13
 */
@Slf4j
@EqualsAndHashCode
@Component
@ServerEndpoint(value = "/user/websocket/{userId}")
public class UserWebsocket {
    /**
     * 静态变量，用来记录当前在线连接数。
     */
    private static AtomicInteger count = new AtomicInteger(0);
    private static Map<Integer, UserWebsocket> userWebsocketMap = new ConcurrentHashMap<>();
    private static Map<Integer, UserWebsocket> serviceWebsocketMap = new ConcurrentHashMap<>();
    private static Map<Integer, SysUser> userMap = new ConcurrentHashMap<>();

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    @Getter
    private Session session;

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public synchronized void onOpen(Session session, @PathParam(value = "userId") Integer userId) {
        // 通过权限系统获取用户的信息
        SysUser sysUser = getSysUser(userId);
        log.info("新客户端连入，用户：{}，session：{}", sysUser.toString(), session.toString());
        // 判断此用户是否已存在；不存在则 在线数 +1
        if (userWebsocketMap.get(userId) == null && serviceWebsocketMap.get(userId) == null) {
            count.incrementAndGet();
        }
        this.session = session;
        // 添加到 map
        if (UserTypeEnum.CUSTOM.getCode().equals(sysUser.getType())) {
            userWebsocketMap.put(sysUser.getId(), this);
        } else {
            serviceWebsocketMap.put(sysUser.getId(), this);
        }
        userMap.put(sysUser.getId(), sysUser);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public synchronized void onClose(@PathParam(value = "userId") Integer userId) {
        // 应该从权限系统获取用户信息
        log.info("一个客户端关闭连接：{}", userId);
        // 从 map 中删除
        SysUser user = userMap.remove(userId);
        if (user != null) {
            if (UserTypeEnum.CUSTOM.getCode().equals(user.getType())) {
                userWebsocketMap.remove(userId);
            } else {
                serviceWebsocketMap.remove(userId);
            }
        }
        // 在线数减1
        count.decrementAndGet();
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息；包含 receiver, content； json格式
     */
    @OnMessage
    public synchronized void onMessage(String message) {
        log.info("用户发送过来的消息为：{}", message);
        BizMessage bizMessage = JsonMapper.defaultMapper().fromJson(message, BizMessage.class);
        if (bizMessage == null || bizMessage.getContent() == null) {
            throw new RuntimeException("");
        }
        if (bizMessage.getReceiver() == null) {
            // 发送给客服
            ArrayList<Integer> userIdList = new ArrayList<>(serviceWebsocketMap.keySet());
            if (userIdList.size() == 0) {
                log.error("客服未上线");
                return;
            }
            // 随机客服
            int index = new Random().nextInt(userIdList.size());
            Integer userId = userIdList.get(index);
            UserWebsocket.sendMessage(bizMessage.getContent(), userId);
            return;
        }
        // 发送给用户
        UserWebsocket websocket = userWebsocketMap.get(bizMessage.getReceiver());
        if (websocket == null) {
            log.error("用户未上线：{}", message);
        }
        UserWebsocket.sendMessage(bizMessage.getContent(), bizMessage.getReceiver());
    }

    /**
     * 发生错误时调用
     */
    @OnError
    public void onError(Throwable error) {
        log.error("websocket出现错误", error);
    }

    /**
     * 定时群发
     */
    @Scheduled(fixedRate = 10000)
    public void productExpire() {
        int index = (int) (Math.random() * STR.length);
        UserWebsocket.sendMessage(LocalDateTime.now().toString() + ":::---->" + STR[index]);
    }

    /**
     * 给指定用户发消息
     *
     * @param message 消息
     * @param userIds 指定用户
     */
    private static void sendMessage(String message, Integer... userIds) {
        HashSet<Integer> userIdSet = new HashSet<>(Arrays.asList(userIds));
        Set<UserWebsocket> userWebsocketSet = userWebsocketMap.entrySet().stream()
                .filter(obj -> userIdSet.contains(obj.getKey())).map(Map.Entry::getValue).collect(Collectors.toSet());
        Set<UserWebsocket> serviceWebsocketSet = serviceWebsocketMap.entrySet().stream()
                .filter(obj -> userIdSet.contains(obj.getKey())).map(Map.Entry::getValue).collect(Collectors.toSet());
        userWebsocketSet.addAll(serviceWebsocketSet);
        // 发消息
        log.info("发消息：{}，发送对象：{}", message, userIdSet);
        userWebsocketSet.forEach(obj -> obj.getSession().getAsyncRemote().sendText(message));
    }

    /**
     * 群发消息
     */
    private static void sendMessage(String message) {
        Set<UserWebsocket> websocketSet = new HashSet<>();
        websocketSet.addAll(userWebsocketMap.values());
        websocketSet.addAll(serviceWebsocketMap.values());
        log.info("群发消息：{}", message);
        websocketSet.forEach(obj -> obj.getSession().getAsyncRemote().sendText(message));
    }

    private static final String[] STR = {"Test随机消息 ：10", "Test随机消息 ：20", "Test随机消息 ：30",
            "Test随机消息 ：40", "Test随机消息 ：50", "Test随机消息 ：60", "Test随机消息 ：70",
            "Test随机消息 ：80", "Test随机消息 ：90", "Test随机消息 ：100"};

    private SysUser getSysUser(Integer userId) {
        if (userId == null) {
            if (userMap.size() == 0) {
                userId = 1;
            } else {
                List<Integer> userIdList =
                        userMap.keySet().stream().sorted(Comparator.comparingInt(obj -> obj)).collect(Collectors.toList());
                Integer id = userIdList.get(userIdList.size() - 1);
                userId = id + 10;
            }
        }
        SysUser sysUser = new SysUser();
        sysUser.setId(userId);
        sysUser.setUsername("测试用户" + userId);
        sysUser.setStatus(true);
        int index = new Random().nextInt(2);
        sysUser.setType(UserTypeEnum.values()[index].getCode());
        return sysUser;
    }

}
